package com.group2.edu.realtime.common.util;

import com.group2.edu.realtime.common.constant.Constant;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author 高耀
 * @date 2024/12/14 16:19<p></p>
 */
public class SQLUtil {
    /**
     * 此方法根据传入的 topic 和 消费者组id <p>
     * 返回kafkaConnector需要的配置部分
     */
    public static String getKafkaProperty(String topic, String groupId) {
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS_ID + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }


    /**
     * 此方法用于根据传入的表名返回Hbase的连接器配置部分
     * @param tableName 表名(不包括namespace部分)
     */
    public static String getHbaseProperty(String tableName) {
        return "WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = '" + Constant.HBASE_NAMESPACE + ":" + tableName + "',\n" +
                " 'zookeeper.quorum' = 'hadoop102,hadoop103,hadoop104:2181',\n" +
                " 'lookup.async' = 'true',\n" + //开启异步才能开启缓存
                " 'lookup.cache' = 'PARTIAL',\n" + //开启缓存
                " 'lookup.partial-cache.max-rows' = '200',\n" + //缓存行数
                " 'lookup.partial-cache.expire-after-write' = '1 hour',\n" + //缓存过期时间
                " 'lookup.partial-cache.expire-after-access' = '1 hour'\n" +
                ")";
    }


    /**
     * 此方法用于根据传入的topic返回upsert-kafka的配置部分
     */
    public static String getUpsertKafkaProperty(String topic) {
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS_ID + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }

    /**
     * 该方法用于根据传入的doris表名返回连接器的配置部分
     * @param tableName 表名
     */
    public static String getDorisProperty(String tableName) {
        return "    WITH (\n" +
                "      'connector' = 'doris',\n" +
                "      'fenodes' = '" + Constant.DORIS_FE_NODES + "',\n" +
                "      'table.identifier' = '" + Constant.DORIS_DATABASE  + "." + tableName + "',\n" +
                "      'username' = 'root',\n" +
                "      'password' = 'aaaaaa',\n" +
                "      'sink.enable-2pc' = 'false'" + //关闭两阶段提交
                ")";
    }


    /**
     * Kafka-Source DDL 语句
     *
     * @param topic   数据源主题
     * @param groupId 消费者组
     * @return 拼接好的 Kafka 数据源 DDL 语句
     */
    public static String getKafkaDDL(String topic, String groupId) {

        return " with ('connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS_ID + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'group-offsets', " +
                " 'properties.auto.offset.reset' = 'earliest')";
    }


    /**
     * UpsertKafka-Sink DDL 语句
     *
     * @param topic 输出到 Kafka 的目标主题
     * @return 拼接好的 UpsertKafka-Sink DDL 语句
     */
    public static String getUpsertKafkaDDL(String topic) {

        return "WITH ( " +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS_ID + "', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }

}
